package common

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Row, SparkSession}

/**
  * Created by hunter.coder 涛哥  
  * 2019/5/6 16:34
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description:  生成ui矩阵
  **/
object UIMatrix {

  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)
    val spark = SparkSession.builder().appName("cb_rec").master("local").getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    val df = spark.read.json("G:\\data_shark\\doit_recommender\\src\\test\\java\\cb_rec\\u.profile.dat")

    // 求event_score： 用户对物品的事件得分
    val event_score = df.rdd.map(row => {
      val gid = row.getAs[String]("gid")
      val logtype = row.getAs[String]("logtype")
      val event = row.getAs[Row]("event")
      val pid = event.getAs[String]("pid")
      val score = event.getAs[Long]("score")

      var sc = 0.0
      // 按不同事件类型，给用户访问的商品计算得分
      logtype match {
        case "pv" => sc += 1
        case "add_cart" => sc += 3
        case "rate" => sc += (score - 3)
      }
      (gid, pid, sc)

    }).toDF("gid", "pid", "sc")

    // 统计用户的文本评价情感分析得分
    val rateDf = spark.read.json("G:\\testdata\\comment\\predict")
    rateDf.createTempView("rate")

    val rateSc = spark.sql(
      """
        |
        |select
        |gid,
        |pid,
        |case
        | when  prediction = 0.0 then -2
        | when prediction = 1.0 then  0
        | when prediction = 2.0 then 2
        |end as sc
        |
        |from rate
        |
      """.stripMargin)

    // 将事件得分和评论情感分析得分，综合成总得分：user_profile
    val user_profile =  event_score.union(rateSc).groupBy("gid","pid").agg(sum("sc").as("score"))

    user_profile.coalesce(1).write.parquet("G:\\testdata\\comment\\uimatrix")

    spark.close()

  }

}
